package com.prac.mq.original;

import java.util.UUID;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

/**
 * 生成业务key作为消息查询的唯一值
 *
 * @author H
 */
@Slf4j
public class MqKeyTest {

    @Test
    public void keyProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("key-producer", false);
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        //producer.setMqClientApiTimeout(1000 * 10);
        producer.start();
        //生成业务key
        String key = UUID.randomUUID().toString();
        log.info(key);
        Message message = new Message("key-topic", "orderKey", key, "order信息".getBytes());
        producer.send(message, 1000 * 10);
        log.info("消息投递成功");
        producer.shutdown();
    }

    @Test
    public void keyConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("key-consumer");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        consumer.subscribe("key-topic", "orderKey");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            MessageExt msg = msgs.get(0);
            log.info("当前线程：" + Thread.currentThread().getId() + "===>" + new String(msg.getBody()));
            log.info(msg.getKeys());
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }


}
